package ProcessFunctionTest

import Source.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 十秒钟内温度连续上升的话就发出警报
 * 当数据上升时注册一个定时器，判断十秒内数据是否是全部上升的
 * 该如何判断十秒内的数据全部上升？
 * 保留前一秒的数据，然后与当前数据比较，现在的温度必须大于前面的温度
 */
object TimeServer {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)


    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    val warningStream = dataStream
      .keyBy(_.id)
      .process(new TempIncreWarning(5000l))

    warningStream.print()
    env.execute()
  }
}

class TempIncreWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
  //定义状态，保留上一个温度值
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(
    new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  //保存定时器的时间戳用于删除
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(
    new ValueStateDescriptor[Long]("ts", classOf[Long]))

  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context,
                              collector: Collector[String]): Unit = {
    //先取出状态
    val lastTemp = lastTempState.value()
    val timerTs = timerTsState.value()

    /**
     * 判断当前温度和上次温度的大小
     * 只有当温度上升，时间状态为0的时候才注册定时器（时间状态为0只有两种可能：一：第一条数据的输入，二：温度不满足连续升温的条件）
     *
     */
    if (i.temperature > lastTemp && timerTs == 0) {
      val ts = context.timerService().currentProcessingTime() + interval //获取当前时间加上时间间隔参数的时间
      context.timerService().registerProcessingTimeTimer(ts) //注册一个时间定时器
      timerTsState.update(ts) //更新时间状态的数据
    } else if (i.temperature < lastTemp) {
      //温度下降，删除定时器
      context.timerService().deleteProcessingTimeTimer(timerTs)
      //更新状态
      timerTsState.clear()
    }
    lastTempState.update(i.temperature) //更新温度
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    out.collect("传感器" + ctx.getCurrentKey + "的温度连续" + interval / 1000 + "秒连续上升")
    //清空时间戳状态
    timerTsState.clear()
  }
}